package com.flink.hbase.sql2hdfs.config;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.Map;

/**
 * HBase 连接配置管理
 * 支持 Kerberos 认证和各种连接参数配置
 */
public class HBaseConnectionConfig implements Serializable {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseConnectionConfig.class);
    private static final long serialVersionUID = 1L;
    
    // HBase 连接参数
    private final String zookeeperQuorum;
    private final String zookeeperPort;
    private final String hbaseZnodeParent;
    private final String hbaseTableName;
    private final String hbaseColumnFamily;
    
    // Kerberos 认证参数
    private final boolean kerberosEnabled;
    private final String krb5ConfPath;
    private final String keytabPath;
    private final String principal;
    private final String hbaseSecurityAuth;
    
    // 连接池和性能参数
    private final int connectionPoolSize;
    private final int maxRetries;
    private final long retryInterval;
    private final long operationTimeout;
    private final long scannerTimeout;
    
    private HBaseConnectionConfig(Builder builder) {
        this.zookeeperQuorum = builder.zookeeperQuorum;
        this.zookeeperPort = builder.zookeeperPort;
        this.hbaseZnodeParent = builder.hbaseZnodeParent;
        this.hbaseTableName = builder.hbaseTableName;
        this.hbaseColumnFamily = builder.hbaseColumnFamily;
        
        this.kerberosEnabled = builder.kerberosEnabled;
        this.krb5ConfPath = builder.krb5ConfPath;
        this.keytabPath = builder.keytabPath;
        this.principal = builder.principal;
        this.hbaseSecurityAuth = builder.hbaseSecurityAuth;
        
        this.connectionPoolSize = builder.connectionPoolSize;
        this.maxRetries = builder.maxRetries;
        this.retryInterval = builder.retryInterval;
        this.operationTimeout = builder.operationTimeout;
        this.scannerTimeout = builder.scannerTimeout;
    }
    
    /**
     * 创建 HBase Configuration
     */
    public Configuration createHBaseConfiguration() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        
        // 基础连接配置
        conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
        conf.set("hbase.zookeeper.property.clientPort", zookeeperPort);
        conf.set("zookeeper.znode.parent", hbaseZnodeParent);
        
        // 性能配置
        conf.setInt("hbase.client.connection.pool.size", connectionPoolSize);
        conf.setInt("hbase.client.retries.number", maxRetries);
        conf.setLong("hbase.client.pause", retryInterval);
        conf.setLong("hbase.client.operation.timeout", operationTimeout);
        conf.setLong("hbase.client.scanner.timeout.period", scannerTimeout);
        
        // 优化配置
        conf.setBoolean("hbase.cluster.distributed", true);
        conf.setInt("hbase.client.write.buffer", 2097152); // 2MB
        conf.setBoolean("hbase.client.scanner.caching.enabled", true);
        conf.setInt("hbase.client.scanner.caching", 1000);
        
        // Kerberos 认证配置
        if (kerberosEnabled) {
            setupKerberosAuthentication(conf);
        }
        
        LOG.info("HBase configuration created with ZK quorum: {}, Kerberos enabled: {}", 
                zookeeperQuorum, kerberosEnabled);
        
        return conf;
    }
    
    /**
     * 设置 Kerberos 认证
     */
    private void setupKerberosAuthentication(Configuration conf) throws IOException {
        LOG.info("Setting up Kerberos authentication...");
        
        // 验证 Kerberos 配置文件
        if (krb5ConfPath != null && !krb5ConfPath.isEmpty()) {
            File krb5File = new File(krb5ConfPath);
            if (!krb5File.exists()) {
                throw new IOException("Kerberos config file not found: " + krb5ConfPath);
            }
            System.setProperty("java.security.krb5.conf", krb5ConfPath);
            LOG.info("Kerberos config file: {}", krb5ConfPath);
        }
        
        // 验证 keytab 文件
        if (keytabPath != null && !keytabPath.isEmpty()) {
            File keytabFile = new File(keytabPath);
            if (!keytabFile.exists()) {
                throw new IOException("Keytab file not found: " + keytabPath);
            }
            LOG.info("Keytab file: {}", keytabPath);
        }
        
        // 设置 HBase 安全认证
        conf.set("hbase.security.authentication", hbaseSecurityAuth);
        conf.set("hadoop.security.authentication", "kerberos");
        
        // 设置 Kerberos 相关配置
        if (principal != null && !principal.isEmpty()) {
            conf.set("hbase.security.auth.enable", "true");
            conf.set("hbase.master.kerberos.principal", principal);
            conf.set("hbase.regionserver.kerberos.principal", principal);
        }
        
        // 启用安全认证
        UserGroupInformation.setConfiguration(conf);
        
        // 如果提供了 keytab 和 principal，进行登录
        if (keytabPath != null && !keytabPath.isEmpty() && 
            principal != null && !principal.isEmpty()) {
            
            LOG.info("Logging in with principal: {} and keytab: {}", principal, keytabPath);
            UserGroupInformation.loginUserFromKeytab(principal, keytabPath);
            LOG.info("Kerberos authentication successful");
        }
    }
    
    /**
     * 验证配置参数
     */
    public void validate() {
        if (zookeeperQuorum == null || zookeeperQuorum.trim().isEmpty()) {
            throw new IllegalArgumentException("Zookeeper quorum cannot be null or empty");
        }
        
        if (hbaseTableName == null || hbaseTableName.trim().isEmpty()) {
            throw new IllegalArgumentException("HBase table name cannot be null or empty");
        }
        
        if (hbaseColumnFamily == null || hbaseColumnFamily.trim().isEmpty()) {
            throw new IllegalArgumentException("HBase column family cannot be null or empty");
        }
        
        if (kerberosEnabled) {
            if (principal == null || principal.trim().isEmpty()) {
                throw new IllegalArgumentException("Principal is required when Kerberos is enabled");
            }
            if (keytabPath == null || keytabPath.trim().isEmpty()) {
                throw new IllegalArgumentException("Keytab path is required when Kerberos is enabled");
            }
        }
    }
    
    // Getter 方法
    public String getZookeeperQuorum() { return zookeeperQuorum; }
    public String getZookeeperPort() { return zookeeperPort; }
    public String getHbaseZnodeParent() { return hbaseZnodeParent; }
    public String getHbaseTableName() { return hbaseTableName; }
    public String getHbaseColumnFamily() { return hbaseColumnFamily; }
    public boolean isKerberosEnabled() { return kerberosEnabled; }
    public String getKrb5ConfPath() { return krb5ConfPath; }
    public String getKeytabPath() { return keytabPath; }
    public String getPrincipal() { return principal; }
    public String getHbaseSecurityAuth() { return hbaseSecurityAuth; }
    public int getConnectionPoolSize() { return connectionPoolSize; }
    public int getMaxRetries() { return maxRetries; }
    public long getRetryInterval() { return retryInterval; }
    public long getOperationTimeout() { return operationTimeout; }
    public long getScannerTimeout() { return scannerTimeout; }
    
    /**
     * 构建器
     */
    public static class Builder {
        // 必需参数
        private String zookeeperQuorum;
        private String hbaseTableName;
        private String hbaseColumnFamily;
        
        // 可选参数 - HBase 连接
        private String zookeeperPort = "2181";
        private String hbaseZnodeParent = "/hbase";
        
        // 可选参数 - Kerberos 认证
        private boolean kerberosEnabled = false;
        private String krb5ConfPath;
        private String keytabPath;
        private String principal;
        private String hbaseSecurityAuth = "kerberos";
        
        // 可选参数 - 性能配置
        private int connectionPoolSize = 10;
        private int maxRetries = 3;
        private long retryInterval = 1000;
        private long operationTimeout = 60000;
        private long scannerTimeout = 60000;
        
        public Builder(String zookeeperQuorum, String hbaseTableName, String hbaseColumnFamily) {
            this.zookeeperQuorum = zookeeperQuorum;
            this.hbaseTableName = hbaseTableName;
            this.hbaseColumnFamily = hbaseColumnFamily;
        }
        
        public Builder zookeeperPort(String zookeeperPort) {
            this.zookeeperPort = zookeeperPort;
            return this;
        }
        
        public Builder hbaseZnodeParent(String hbaseZnodeParent) {
            this.hbaseZnodeParent = hbaseZnodeParent;
            return this;
        }
        
        public Builder enableKerberos(String principal, String keytabPath) {
            this.kerberosEnabled = true;
            this.principal = principal;
            this.keytabPath = keytabPath;
            return this;
        }
        
        public Builder krb5ConfPath(String krb5ConfPath) {
            this.krb5ConfPath = krb5ConfPath;
            return this;
        }
        
        public Builder hbaseSecurityAuth(String hbaseSecurityAuth) {
            this.hbaseSecurityAuth = hbaseSecurityAuth;
            return this;
        }
        
        public Builder connectionPoolSize(int connectionPoolSize) {
            this.connectionPoolSize = connectionPoolSize;
            return this;
        }
        
        public Builder maxRetries(int maxRetries) {
            this.maxRetries = maxRetries;
            return this;
        }
        
        public Builder retryInterval(long retryInterval) {
            this.retryInterval = retryInterval;
            return this;
        }
        
        public Builder operationTimeout(long operationTimeout) {
            this.operationTimeout = operationTimeout;
            return this;
        }
        
        public Builder scannerTimeout(long scannerTimeout) {
            this.scannerTimeout = scannerTimeout;
            return this;
        }
        
        public HBaseConnectionConfig build() {
            HBaseConnectionConfig config = new HBaseConnectionConfig(this);
            config.validate();
            return config;
        }
    }
    
    /**
     * 从 Map 创建配置
     */
    public static HBaseConnectionConfig fromMap(Map<String, Object> configMap) {
        Builder builder = new Builder(
            (String) configMap.get("zookeeper.quorum"),
            (String) configMap.get("hbase.table.name"),
            (String) configMap.get("hbase.column.family")
        );
        
        // 可选参数
        if (configMap.containsKey("zookeeper.port")) {
            builder.zookeeperPort((String) configMap.get("zookeeper.port"));
        }
        if (configMap.containsKey("hbase.znode.parent")) {
            builder.hbaseZnodeParent((String) configMap.get("hbase.znode.parent"));
        }
        
        // Kerberos 参数
        Boolean kerberosEnabled = (Boolean) configMap.get("kerberos.enabled");
        if (kerberosEnabled != null && kerberosEnabled) {
            String principal = (String) configMap.get("kerberos.principal");
            String keytabPath = (String) configMap.get("kerberos.keytab.path");
            builder.enableKerberos(principal, keytabPath);
            
            if (configMap.containsKey("kerberos.krb5.conf.path")) {
                builder.krb5ConfPath((String) configMap.get("kerberos.krb5.conf.path"));
            }
        }
        
        // 性能参数
        if (configMap.containsKey("connection.pool.size")) {
            builder.connectionPoolSize((Integer) configMap.get("connection.pool.size"));
        }
        if (configMap.containsKey("max.retries")) {
            builder.maxRetries((Integer) configMap.get("max.retries"));
        }
        if (configMap.containsKey("retry.interval")) {
            builder.retryInterval((Long) configMap.get("retry.interval"));
        }
        if (configMap.containsKey("operation.timeout")) {
            builder.operationTimeout((Long) configMap.get("operation.timeout"));
        }
        if (configMap.containsKey("scanner.timeout")) {
            builder.scannerTimeout((Long) configMap.get("scanner.timeout"));
        }
        
        return builder.build();
    }
    
    @Override
    public String toString() {
        return "HBaseConnectionConfig{" +
                "zookeeperQuorum='" + zookeeperQuorum + '\'' +
                ", zookeeperPort='" + zookeeperPort + '\'' +
                ", hbaseTableName='" + hbaseTableName + '\'' +
                ", hbaseColumnFamily='" + hbaseColumnFamily + '\'' +
                ", kerberosEnabled=" + kerberosEnabled +
                ", principal='" + principal + '\'' +
                ", connectionPoolSize=" + connectionPoolSize +
                ", maxRetries=" + maxRetries +
                '}';
    }
} 